package com.atguigu.tingshu.album.service.impl;

import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackInfoMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.base.BaseEntity;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
//@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImplWhitRedisson extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;
	@Autowired
	private AlbumStatMapper albumStatMapper;
	@Autowired
	private AlbumAttributeValueMapper albumAttributeValueMapper;
	@Autowired
	private AlbumAttributeValueService albumAttributeValueService;
	@Autowired
	private KafkaService kafkaService;
	@Autowired
	private TrackInfoMapper trackInfoMapper;
	@Autowired
	private RedissonClient redissonClient;
	@Autowired
	private RedisTemplate redisTemplate;


	@Override
	@Transactional
	public void saveAlbumInfo(AlbumInfoVo albumInfoVo, Long userId) {

		try{
			AlbumInfo albumInfo = new AlbumInfo();

			BeanUtils.copyProperties(albumInfoVo,albumInfo);
			albumInfo.setUserId(userId);
			albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS); //审核通过

			if (!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfoVo.getPayType())) {
				//只要收费就设置前5集为免费试听集数
				albumInfo.setTracksForFree(5);
			}

			// @TableId(type = IdType.AUTO) 插入完成后能够自动获取主键自增的id
			albumInfoMapper.insert(albumInfo);

			saveAlbumAttributeValue(albumInfoVo, albumInfo);

			if ("1".equals(albumInfoVo.getIsOpen())) {
				//上架
				kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER,albumInfo.getId().toString());
			}

			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);
		}catch  (BeansException e) {
			log.error("保存专辑失败");
			throw new RuntimeException(e);
		}
	}

	public void saveAlbumAttributeValue(AlbumInfoVo albumInfoVo, AlbumInfo albumInfo) {
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();

		if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {

			List<AlbumAttributeValue> attributeValueList = albumAttributeValueVoList.stream().map(item -> {

				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
				//	赋值
				albumAttributeValue.setAlbumId(albumInfo.getId());
				albumAttributeValue.setAttributeId(item.getAttributeId());
				albumAttributeValue.setValueId(item.getValueId());
				return albumAttributeValue;

			}).collect(Collectors.toList());

			albumAttributeValueService.saveBatch(attributeValueList);
		}
	}

	private void saveAlbumStat(Long albumId, String statPlay) {

		//	创建统计对象
		AlbumStat albumStat = new AlbumStat();
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statPlay);
		albumStat.setStatNum(0);
		albumStatMapper.insert(albumStat);

	}


	@Override
	public IPage<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pagePage, AlbumInfoQuery query) {

		return albumInfoMapper.selectUserAlbumPage(pagePage,query);
	}



	@Transactional(rollbackFor = Exception.class)
	@Override
	public void removeAlbumInfo(Long albumId) {

		this.removeById(albumId);

		albumStatMapper.delete(new LambdaQueryWrapper<AlbumStat>().eq(AlbumStat::getAlbumId,albumId));

		albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));

	}


	@Override
	public AlbumInfo getAlbumInfoById(Long albumId) {

		//	查询缓存：String set key value;
		String key = RedisConstant.ALBUM_INFO_PREFIX + albumId;

		try {
			//  从缓存中获取数据
			AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);

			//  判断当前专辑对象是否为空   //   为null缓存中没有数据，查询数据库，避免缓存击穿要加分布式锁！
			if (null == albumInfo) {

				//  声明一个锁的key
				String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;

				RLock lock = redissonClient.getLock(lockKey);
				lock.lock();

				try{
					//  获取到锁的时候，最好再查询一次缓存。 因为第一个抢到锁的线程将数据存到了缓存，之后的线程直接查缓存就能拿到数据返回
					albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);
					log.info("查询了缓存....");
					if (null != albumInfo) {
						return albumInfo;
					}

					//查询数据库
					 albumInfo = getAlbumInfoDB(albumId);

					//  判断当前这个对象是否存在.
					 if (albumInfo == null) {
						 //  返回数据  说明数据库也没有 返回一个空对象并设置过期时间
						 this.redisTemplate.opsForValue().set(key, new AlbumInfo(), RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
						 //  返回空对象
						 return new AlbumInfo();
					 }else {
						 //  数据库中有数据
						 this.redisTemplate.opsForValue().set(key, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
						 return albumInfo;
					 }
				}finally {
					lock.unlock();
				}
			}else{  //缓存数据不为空
				return albumInfo;
			}
		} catch (Exception e) {
			log.error("出现了异常{}专辑Id：\t", albumId);
			//  调用sendMsg(phoneNumber);
			e.printStackTrace();
		}
		return null;
	}

	@Nullable
	private AlbumInfo getAlbumInfoDB(Long albumId) {
		AlbumInfo albumInfo = this.lambdaQuery().eq(BaseEntity::getId, albumId).one();

		if (albumInfo != null) {

			List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));

			albumInfo.setAlbumAttributeValueVoList(albumAttributeValues);

			return albumInfo;
		}

		return null;
	}


	@Transactional(rollbackFor = Exception.class)
	@Override
	public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {

		AlbumInfo albumInfo = new AlbumInfo();
		BeanUtils.copyProperties(albumInfoVo,albumInfo);
		albumInfo.setId(albumId);
		albumInfoMapper.updateById(albumInfo);

		//修改album_attribute_value表 先删除 再新增
		albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));

		saveAlbumAttributeValue(albumInfoVo,albumInfo);

		if ("1".equals(albumInfoVo.getIsOpen())) {
			//上架
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER,albumInfo.getId().toString());
		}else {
			//下架
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER,albumInfo.getId().toString());
		}

	}


	@Override
	public List<AlbumInfo> findUserAllAlbumList(Long userId) {

		//	根据用户Id 查询专辑列表
		//	List<AlbumInfo> albumInfoList = albumInfoMapper.selectList(new LambdaQueryWrapper<AlbumInfo>().eq(AlbumInfo::getUserId, userId));
		//	Page<AlbumInfo> albumInfoPage = new Page<>(1,20);
		//	select id, album_title from album_info where user_id = ? order by id desc limit 0,20;

		//		LambdaQueryWrapper<AlbumInfo> wrapper = new LambdaQueryWrapper<AlbumInfo>()
		//				.eq(AlbumInfo::getUserId, userId)
		//				.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle)
		//				.orderByDesc(AlbumInfo::getId);
		//		//	查询数据
		//		Page<AlbumInfo> selectPage = albumInfoMapper.selectPage(albumInfoPage, wrapper);
		//		return selectPage.getRecords();
		//	select id, album_title from album_info where user_id = ? order by id desc limit 20;

		LambdaQueryWrapper<AlbumInfo> wrapper = new LambdaQueryWrapper<AlbumInfo>()
				.eq(AlbumInfo::getUserId, userId)
				.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle)
				.orderByDesc(AlbumInfo::getId)
				.last(" limit 20 ");

		return albumInfoMapper.selectList(wrapper);
	}


	@Override
	public List<AlbumAttributeValue> findAlbumAttributeValueByAlbumId(Long albumId) {

		return albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));
	}


	@Override
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		return albumInfoMapper.selectAlbumStatVo(albumId);
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void trackStatUpdate(TrackStatMqVo trackStatMqVo) throws Exception{
		// 更新声音数据
		trackInfoMapper.trackStatUpdate(trackStatMqVo.getTrackId(),trackStatMqVo.getStatType(),trackStatMqVo.getCount());
		//跟新专辑
		if (SystemConstant.TRACK_STAT_PLAY.equals(trackStatMqVo.getStatType())) {
			albumInfoMapper.updateAlbumStat(trackStatMqVo.getAlbumId(),SystemConstant.ALBUM_STAT_PLAY,trackStatMqVo.getCount());
		}
	}
}
